 1.Kafka高级特性解析之主题中的管理
   
   使用kafka-topics.sh脚本：
   选项                     说明
   --config                 为创建的或修改的主题指定配置信息。支持持下述配置条目：
   <String: name=value>     cleanup.policy、compression.type、delete.retention.ms
                            、file.delete.delay.ms、flush.messages
                            、flush.ms、follower.replication.throttled.replicas
                            、index.interval.bytes、leader.replication.throttled.replicas
                            max.message.bytes、message.format.version
                            、message.timestamp.difference.max.ms、message.timestamp.type
                            、min.cleanable.dirty.ratio、min.compaction.lag.ms
                            、min.insync.replicas、preallocate、retention.bytes
                            、retention.ms、segment.bytes、segment.index.bytes、
							segment.jitter.ms、segment.ms、unclean.leader.election.enable
   --create                 创建一个新主题
   --delete                 删除一个主题
   --delete-config          删除现有主题的一个主题配置条目。这些条目就是在--
   <String: name>           config 中给出的配置条目。
   --alter                  更改主题的分区数量，副本分配和/或配置条目。
   -describe                列出给定主题的细节。
   --disable-rack-aware     禁用副本分配的机架感知。
   --force                  抑制控制台提示信息
   --help                   打印帮助信息
   --if-exists              如果指定了该选项，则在修改或删除主题的时候，
                            只有主题存在才可以执行。
   -if-not-exists           在创建主题的时候，如果指定了该选项，
                            则只有主题不存在的时候才可以执行命令。
   --list                   列出所有可用的主题。
   --partitions             要创建或修改主题的分区数。
  <Integer: # of partitions> 
  --replica-assignment<     当创建或修改主题的时候自动指定
  String:                   partition-to-broker的分配关系。
  broker_id_for_part1_replica1:
  broker_id_for_part1_replica2,
  broker_id_for_part2_replica1:
  broker_id_for_part2_replica2, 
  ...>
  --replication-factor     要创建的主题分区副本数。1表示只有一个副本  
  <Integer:replication factor>  ，也就是 Leader副本。
  --topic <String: topic>  要创建、修改或描述的主题名称。除了创建，
                           修改和描述在这里还可以使用正则表达式。
  --topics-with-overrides  if set when describing topics, only show 
                           topics that have overridden configs
  --unavailable-partitions if set when describing topics, only show 
                           partitions whose leader is not available
  --under-replicated       if set when describing topics, only 
  -partitions              show under replicated partitions
  --zookeeper              必需的参数：连接zookeeper的字符串，逗号分隔的多个
  <String: urls>           host:port列表。多个URL可以故障转移。
  主题中可以使用的参数定义：
  属性                             默认值            服务器默认属性      
  cleanup.policy                   delete           log.cleanup.policy
  说明：要么是”delete“要么是”compact“； 这个字符串指明了针对旧日志部分的利用
⽅式；默认⽅式（"delete"）将会丢弃旧的部分当他们的回收时间或者尺寸限
制到达时。”compact“将会进行日志压缩
  compression.type                 none
  说明：producer⽤于压缩数据的压缩类型。默认是无压缩。正确的选项值是none、
gzip、snappy。压缩最好用于批量处理，批量处理消息越多，压缩性能越好。
  delete.retention.ms         86400000(24hours)     log.cleaner.delete.retention.ms
  说明：对于压缩⽇志保留的最⻓时间，也是客户端消费消息的最长时间，通
log.retention.minutes的区别在于一个控制未压缩数据，一个控制压缩后的
数据。此项配置可以在topic创建时的置顶参数覆盖
  flush.ms                         None             log.flush.interval.ms
  说明:此项配置⽤来置顶强制进行fsync日志到磁盘的时间间隔；例如，如果设置为
1000,那么每1000ms就需要进行一次fsync。一般不建议使用这个选项
  flush.messages                   None             log.flush.interval.messages
  说明：此项配置指定时间间隔：强制进行fsync日志。例如，如果这个选项设置为1，
一次fsync。⼀般来说，建议你不要设置这个值。此参数的设置,需要在"数据可靠性"
与"性能"之间做必要的权衡.如果此值过大,将会导致每次"fsync"的时间
较长(IO阻塞),如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的
client请求有一定的延迟.物理server故障,将会导致没有fsync的消息丢失.
  index.interval.bytes             4096             log.index.interval.bytes
  说明：默认设置保证了我们每4096个字节就对消息添加一个索引，更多的索引使得
阅读的消息更加靠近，但是索引规模却会由此增大；一般不需要改变这个选项
  max.message.bytes              1000000            max.message.bytes
  说明：kafka追加消息的最大尺寸。注意如果你增大这个尺寸，你也必须增大你
consumer的fetch 尺寸，这样consumer才能fetch到这些最大尺寸的消息。
  min.cleanable.dirty.ratio        0.5              min.cleanable.dirty.ratio
  说明：此项配置控制log压缩器试图进行清除日志的频率。默认情况下，将避免清除
压缩率超过50%的日志。这个比率避免了最⼤的空间浪费
  min.insync.replicas               1               min.insync.replicas
  说明：当producer设置request.required.acks为-1 时,min.insync.replicas指定
replicas的最⼩数⽬（必须确认每一个repica的写数据都是成功的），如果这
个数目没有达到，producer会产生异常。
  retention.bytes                 None              log.retention.bytes
  说明：如果使用“delete”的retention 策略，这项配置就是指在删除日志之前，日志
所能达到的最大尺寸。默认情况下，没有尺寸限制而只有时间限制
  retention.ms                      7               days log.retention.minutes
  说明：如果使用“delete”的retention策略，这项配置就是指删除日志前日志保存的时
间。
  segment.bytes                    1GB              log.segment.bytes
  说明：kafka中log日志是分成一块块存储的，此配置是指log日志划分成块的大小
  segment.index.bytes             10MB              log.index.size.max.bytes
  说明：此配置是有关offsets和⽂件位置之间映射的索引文件的大小；一般不需要修
改这个配置
  segment.jitter.ms                 0               log.roll.jitter.{ms,hours}
  说明：The maximum jitter to subtract from logRollTimeMillis.
  segment.ms                     7 days             log.roll.hours
  说明：即使log的分块文件件没有达到需要删除、压缩的大小，一旦log 的时间达到这
个上限，就会强制新建一个log分块文件
  unclean.leader.election.enable true
  说明：指明了是否能够使不在ISR中replicas设置用来作为leader
 
 2.创建主题
   
   kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic tp_test_01
--partitions 3 --replication-factor 1
   kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic tp_test_02
--partitions 3 --replication-factor 1 --config cleanup.policy=compact
   kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic tp_test_03
--partitions 5 --replication-factor 1 --config compression.type=gzip --config 
max.message.bytes=512
   kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test_02 --
partitions 3 --replication-factor 1 --config max.message.bytes=1048576 --config
segment.bytes=10485760
 3.查看主题
   
   kafka-topics.sh --zookeeper linux121:2181/myKafka --list
   kafka-topics.sh --zookeeper linux121:2181/myKafka --describe --topic tp_test_01
   kafka-topics.sh --zookeeper linux121:2181/myKafka --describe --topics-with-overrides
 4.修改主题
   
   kafka-topics.sh --zookeeper linux121:2181/myKafka --alter --topic tp_test_02
--config segment.bytes=104857600
   kafka-topics.sh --zookeeper linux121:2181/myKafka --alter --delete-config
max.message.bytes --topic tp_test_03
   
 5.删除主题
   kafka-topics.sh --zookeeper linux121:2181/myKafka --delete --topic tp_test_03
   
   给主题添加删除的标记：
   要过一段时间删除.
 
 6.增加分区
   
   通过命令行工具操作，主题的分区只能增加，不能减少。否则报错：
   ERROR org.apache.kafka.common.errors.InvalidPartitionsException: The number of
partitions for a topic can only be increased. Topic tp_demo_001 currently has 2 partitions,
1 would not be an increase.
   通过--alter修改主题的分区数，增加分区。
   kafka-topics.sh --zookeeper linux121:2181/myKafka --alter --topic tp_demo_001 
--partitions 3
 
 7.分区副本的分配-了解
   
   副本分配的三个目标：
   1).均衡地将副本分散于各个broker上
   2).对于某个broker上分配的分区，它的其他副本在其他broker上
   3).如果所有的broker都有机架信息，尽量将分区的各个副本分配到不同机架上的broker。
   在不考虑机架信息的情况下：
   1).第一个副本分区通过轮询的方式挑选一个broker，进行分配。该轮询从broker列表的随机位置进行轮询。
   2).其余副本通过增加偏移进行分配。
   分配案例：
   broker-0 broker-1 broker-2 broker-3 broker-4
    p0        p1       p2        p3       p4    (1st replica)
    p5        p6       p7        p8       p9    (1st replica)
    p4        p0       p1        p2       p3    (2nd replica)
    p8        p9       p5        p6       p7    (2nd replica)
    p3        p4       p0        p1       p2    (3nd replica)
    p7        p8       p9        p5       p6    (3nd replica)
   考虑到机架信息，首先为每个机架创建⼀个broker列表。如：
   三个机架（rack1，rack2，rack3），六个broker（0，1，2，3，4，5）
   brokerID -> rack
      0 -> "rack1", 1 -> "rack3", 2 -> "rack3", 3 -> "rack2", 4 -> "rack2", 5 -> "rack1"
   rack1：0，5
   rack2：3，4
   rack3：1，2
   这broker列表为rack1的0，rack2的3，rack3的1，rack1的5，rack2的4，rack3的2
   即：0, 3, 1, 5, 4, 2
   通过简单的轮询将分区分配给不同机架上的broker：
   每个分区副本在分配的时候在上一个分区第一个副本开始分配的位置右移一位。
   六个broker，六个分区，正好最后一个分区的第⼀个副本分配的位置是该broker列表的最
   如果有更多的分区需要分配，则算法开始对follower副本进行移位分配。
   这主要是为了避免每次都得到相同的分配序列。
   此时，如果有一个分区等待分配（分区6），这按照如下方式分配：
   6 -> 0,4,2 (而不是像分区0那样重复0,3,1)
   跟机架相关的副本分配中，永远在机架相关的broker列表中轮询地分配第一个副本。
其余的副本，倾向于机架上没有副本的broker进行副本分配，除非每个机架有一个副本。
然后其他的副本⼜通过轮询的方式分配给broker。
   
   结果是，如果副本的个数大于等于机架数，保证每个机架最少有一个副本。
否则每个机架最多保有一个副本。
   如果副本的个数和机架的个数相同，并且每个机架包含相同个数的broker，可以保证副本在机架和
broker之间均匀分布。
    [zk: localhost:2181(CONNECTED) 0] get /myKafka/brokers/topics/tp_test_02/
partitions/0/state
    {"controller_epoch":11,"leader":0,"version":1,"leader_epoch":0,"isr":[0]}
   上图，tp_eagle_01主题的分区0分配信息：leader分区在broker1上，同步副本分区是1和2，也就是在broker1和
broker2上的两个副本分区是同步副本分区，其中一个是leader分区。
   